package com.atguigu.flink.datastream.source;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 10:37
 * <p>
 * 自定义数据源
 * <p>
 * 基于旧的API
 * env.addSource(SourceFunction)
 */
public class Flink04_UserDefineSource {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //使用自定义Source
        ClickSource clickSource = new ClickSource();
        DataStreamSource<Event> ds = env.addSource(clickSource);
        ds.print();

        //开启新的线程控制开关
        new Thread(
                new Runnable() {
                    private Boolean isRunning = true;

                    @Override
                    public void run() {
                        while (isRunning) {
                            // 判断用户的意图(是否要关)
                            // 借助于外部的设备(  hdfs 、 mysql 、 file 等)来控制
                            try {
                                FileSystem fileSystem = FileSystem.get(
                                        new URI("hdfs://hadoop102:8020")
                                        ,
                                        new Configuration()
                                        ,
                                        "atguigu"
                                );

                                //判断是否存在cancel目录
                                boolean exists = fileSystem.exists(new Path("hdfs://hadoop102:8020/cancel"));
                                if (exists) {
                                    //停止数据生成
                                    clickSource.cancel();
                                    //停止当前线程
                                    isRunning = false;
                                }

                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }
                }
        ).start();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}


